package realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import realtime.util.MyKafkaUtil;

/**
 * @author MengX
 * @create 2023/3/25 09:31:50
 */
public class DwdFavorInfo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String groupID = "dwd_favor_info";
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL(groupID));

        Table favorInfo = tableEnv.sqlQuery("" +
                "select  \n" +
                "   data['id'] id,\n" +
                "   data['course_id'] course_id,\n" +
                "   data['user_id'] user_id,\n" +
                "   data['create_time'] create_time,\n" +
                "   data['update_time'] update_time,\n" +
                "   data['deleted'] deleted\n" +
                "from `topic_db`\n" +
                "where `table` = 'favor_info' and `type` = 'insert'");

        tableEnv.createTemporaryView("favor_info",favorInfo);

        tableEnv.executeSql("" +
                "CREATE TABLE `dwd_favor_info` (\n" +
                "  `id` STRING,\n" +
                "  `course_id` STRING,\n" +
                "  `user_id` STRING,\n" +
                "  `create_time`STRING,\n" +
                "  `update_time` STRING,\n" +
                "  `deleted` STRING\n" +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_favor_info"));

        tableEnv.executeSql("insert into dwd_favor_info select * from favor_info");
    }
}
